package com.liao.mr;

import com.liao.converter.impl.DimensionConverter;
import com.liao.kv.key.ComDimension;
import com.liao.kv.value.CountDurationValue;
import com.liao.kv.base.BaseDimension;
import com.liao.kv.base.BaseValue;
import com.liao.util.JDBCCacheBean;
import com.liao.util.JDBCUtil;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/*传到mysql需要的格式类*/
public class MySQLOutputFormat extends OutputFormat<BaseDimension, BaseValue> {

    @Override //靠的就是这个函数输出到MySQL
    public RecordWriter<BaseDimension, BaseValue> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException {
        //创建jdbc连接
        Connection conn = null;
        try {
            conn = JDBCCacheBean.getInstance();//单例获取Connection
            //关闭自动提交，以便于批量提交
            conn.setAutoCommit(false);
        } catch (SQLException e) {
            throw new IOException(e);
        }
        return new MysqlRecordWriter(conn);
    }

    @Override
    public void checkOutputSpecs(JobContext jobContext) throws IOException, InterruptedException {
        // 校检输出
    }

    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        String name = taskAttemptContext.getConfiguration().get(FileOutputFormat.OUTDIR);
        Path output = name == null ? null : new Path(name);
        return new FileOutputCommitter(output, taskAttemptContext);
    }

    /*内部静态类*/
    static class MysqlRecordWriter extends RecordWriter<BaseDimension, BaseValue> {

        private Connection conn = null;
        private PreparedStatement preparedStatement = null;
        private int batchNumber = 500;//缓存sql条数的边界条数
        int count = 0;//用于计数客户端已经缓存的条数

        public MysqlRecordWriter(Connection conn) {
            this.conn = conn;
        }

        @Override  //write方法可能会有多线程调用，key会有5个数据：姓名 电话 年 月 日 value有两个数据：某维度的通话次数 某维度通话时间总和
        public void write(BaseDimension key, BaseValue v) throws IOException{//插入数据到Mysql,一个电话有365+12+1的数据，每天每月每年
            ComDimension comDimension=(ComDimension) key;
            CountDurationValue value=(CountDurationValue) v;
            /*`id_date_contact`, `id_date_dimension`, `id_contact`, `call_sum`, `call_duration_sum`   后面两列字段是value里有的`call_sum`, `call_duration_sum`*/
            String sql = "INSERT INTO `tb_call`(`id_date_contact`, `id_date_dimension`, `id_contact`, `call_sum`, `call_duration_sum`) VALUES(?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE `id_date_contact` = ? ;";
            /*维度转换*/
            /*id_date_contact`, `id_date_dimension`, `id_contact`, 第一列数据是后面第二第三列数据拼接的，而第二第三列数据是封装在不同的维度表里*/
            DimensionConverter converter = new DimensionConverter();//所以我们整了一个维度转换器
            // 将姓名和电话转换成它们对应的id
            int contactId = converter.getDimensionId(comDimension.getContactDimension());//*因为write方法可能会有多线程调用，可能getDimensionId方法也会有多线程调用
            //将年月日转换成它们所对应的id
            int dateId = converter.getDimensionId(comDimension.getDateDimension());//*因为write方法可能会有多线程调用，可能getDimensionId方法也会有多线程调用
            //拼接成第一个列
            String date_contact = dateId + "_" + contactId;
            //通话次数
            int call_sum = value.getCallSum();
            //通话时长
            int call_duration_sum = value.getCallDurationSum();
            /*preparestatement填值，批量SQL处理*/
            try {
                if (preparedStatement == null)
                    preparedStatement = conn.prepareStatement(sql);
                int i = 0;//用于填充preparedStatement的参数索引
                preparedStatement.setString(++i, date_contact);//id_date_contact
                preparedStatement.setInt(++i, dateId);//id_date_dimension
                preparedStatement.setInt(++i, contactId);//id_contact
                preparedStatement.setInt(++i, call_sum);//call_sum
                preparedStatement.setInt(++i, call_duration_sum);//call_duration_sum

                preparedStatement.setString(++i, date_contact);
                preparedStatement.addBatch();//将SQL缓存到客户端
                //当前缓存了多少个sql语句等待批量执行，计数器
                count++;
                // 批量提交
                if (count >= batchNumber) {
                    preparedStatement.executeBatch(); // 批量提交
                    conn.commit(); // 因为之前设置的是不自动提交，所以这里提交一下事务
                    count = 0;
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }

        @Override //等所有write方法执行完，系统自动调用该函数进行关闭
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            try {
                preparedStatement.executeBatch();// 批量提交
                conn.commit();//那个如果积累只有400+的条数，判断条件要大于500才提交的话，所以再关闭前需要再提交一次
            } catch (SQLException e) {
                e.printStackTrace();
            }finally {
                JDBCUtil.close(conn, preparedStatement, null);
            }
        }
    }
}
